package com.ehualu.javahadoop.test;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hdfs.DistributedFileSystem;

import java.io.IOException;
import java.net.URI;


/**
 * @Author: 吴敬超
 * @Date: 2020/3/2 13:05
 */
public class javahdfs {
    private final static Log LOG = LogFactory.getLog(javahdfs.class.getName());
    private static String PATH_TO_HDFS_SITE_XML = javahdfs.class.getClassLoader().getResource("hdfs-site.xml")
            .getPath();
    private static String PATH_TO_CORE_SITE_XML = javahdfs.class.getClassLoader().getResource("core-site.xml")
            .getPath();

    private static Configuration conf = null;
    private FileSystem fSystem; /* HDFS file system */
    private String DEST_PATH;
    private String FILE_NAME;

    private void instanceBuild() throws IOException {
        // get filesystem
        // 一般情况下，FileSystem对象JVM里唯一，是线程安全的，这个实例可以一直用，不需要立马close。
        // 注意：
        // 若需要长期占用一个FileSystem对象的场景，可以给这个线程专门new一个FileSystem对象，但要注意资源管理，别导致泄露。
        // 在此之前，需要先给conf加上：
        // conf.setBoolean("fs.hdfs.impl.disable.cache",
        // true);//表示重新new一个连接实例，不用缓存中的对象。
        fSystem = FileSystem.get(conf);
    }

    private static void confLoad() throws IOException {

        conf = new Configuration();
        // conf file
        conf.addResource(new Path(PATH_TO_HDFS_SITE_XML));
        conf.addResource(new Path(PATH_TO_CORE_SITE_XML));
        System.setProperty("HADOOP_USER_NAME", "hdfs");
        // conf.addResource(new Path(PATH_TO_SMALL_SITE_XML));
    }

    private boolean createPath(final Path filePath) throws IOException {
        if (!fSystem.exists(filePath)) {
            fSystem.mkdirs(filePath);
//            fSystem.listFiles();
//            fSystem.listStatus(Path);
//            fSystem.
//            fSystem.initialize();
        }
        return true;
    }


    public javahdfs(String path, String fileName) throws IOException {
        this.DEST_PATH = path;
        this.FILE_NAME = fileName;
        instanceBuild();
    }

    /**
     * create directory
     *
     * @throws IOException
     */
    private void mkdir() throws IOException {
        Path destPath = new Path(DEST_PATH);
        if (!createPath(destPath)) {
//            LOG.error("failed to create destPath " + DEST_PATH);

            System.out.println("failed to create destPath " + DEST_PATH);
            return;
        }

        System.out.println("success to create path " + DEST_PATH);
    }


    public void recoverLease(String path) throws IOException {
        DistributedFileSystem fs = new DistributedFileSystem();
        Configuration conf = new Configuration();
        fs.initialize(URI.create(path), conf);

        boolean fileflag=fs.exists(new Path(path));
        System.out.println(fileflag);

//        fs.recoverLease(new Path(path));
        fs.close();
    }


    public void getname() {

        String uri = "hdfs://neusoft-master:9000/user";
        Configuration conf = new Configuration();
        try {
            FileSystem fs = FileSystem.get(URI.create(uri), conf);
            Path Path = new Path(uri);
            FileStatus[] status = fs.listStatus(Path);
            for (int i = 0; i < status.length; i++) {
                System.out.println(status[i].getPath().toString());
            }
            fs.close();
        } catch (IOException e) {
            e.printStackTrace();
        }

    }
    private void filflag(String path) throws IOException {


        DistributedFileSystem fs = new DistributedFileSystem();

        fs.initialize(URI.create(path), conf);
        FileStatus[] fileStatuses = fs.listStatus(new Path(path));

        for (FileStatus fileStatus:fileStatuses){


            String everypath=fileStatus.getPath().toString();
            System.out.println(everypath);
            Path evepath=new Path(everypath);
            if (fs.isFile(evepath)){
//                System.out.println("是文件修复");
                LOG.info("是文件修复");
                fs.recoverLease(evepath);


            }else {

                LOG.info("是路径忽略");

            }


//            fs.recoverLease(new Path(everypath));

        }


//        fs.recoverLease(new Path(path));
        fs.close();
//        Path destPath = new Path(DEST_PATH);
//
//        if (!fSystem.exists(destPath)) {
//           System.out.println("文件不存在");
////            fSystem.
////            fSystem.initialize();
//        }else {
//            DistributedFileSystem fs = new DistributedFileSystem();
//
//            fs.initialize(URI.create(path), conf);
//            fs.recoverLease(new Path(path));
//            fs.close();
//
//        }

    }


    private void recover(String path) throws IOException {


        DistributedFileSystem fs = new DistributedFileSystem();

        fs.initialize(URI.create(path), conf);
        FileStatus[] fileStatuses = fs.listStatus(new Path(path));

        for (FileStatus fileStatus:fileStatuses){


            String everypath=fileStatus.getPath().toString();
            System.out.println(everypath);
            Path evepath=new Path(everypath);
            if (fs.isFile(evepath)){
//                System.out.println("是文件修复");
                LOG.info("是文件修复");
                fs.recoverLease(evepath);


            }else {

                LOG.info("是路径忽略");

            }


//            fs.recoverLease(new Path(everypath));

        }


//        fs.recoverLease(new Path(path));
        fs.close();
//        Path destPath = new Path(DEST_PATH);
//
//        if (!fSystem.exists(destPath)) {
//           System.out.println("文件不存在");
////            fSystem.
////            fSystem.initialize();
//        }else {
//            DistributedFileSystem fs = new DistributedFileSystem();
//
//            fs.initialize(URI.create(path), conf);
//            fs.recoverLease(new Path(path));
//            fs.close();
//
//        }

    }

    public static void main(String[] args) {


        try {
            confLoad();
            javahdfs jhd = new javahdfs("/hdfstest/tt", "tt.tmp");
//            jhd.mkdir();
//            jhd.recover("hdfs://bigdataha/user/super_rj/hive/textfile/lcgcsj/dt=20190801/qx=371501/gcxx.tmp");
//            jhd.recover("hdfs://bigdataha/user/super_rj/hive/textfile/lcgcsj/dt=20190801/");
            jhd.recoverLease("hdfs://bigdataha/user/super_rj/hive/textfile/lcgcsj/dt=20190801/");

        } catch (IOException e) {
            e.printStackTrace();
        }


    }
}
